package db_test

import (
	"fmt"
	"log"

	"gintest.com/m/util"
	"github.com/rabbitmq/amqp091-go"
)

const MQ_URL = "amqp://guest:guest@localhost:5672/"

// rabbitMQ结构体
type RabbitMQ struct {
	conn         *amqp091.Connection
	ch           *amqp091.Channel
	QueueName    string // 队列名
	ExchangeName string // 交换机名称
	Key          string // bind key 名称
	MqUrl        string // 连接url
}

// 创建结构体实例
func NewRabbitMQStruct(queueName string, exchange string, key string) *RabbitMQ {
	return &RabbitMQ{
		QueueName:    queueName,
		ExchangeName: exchange,
		Key:          key,
		MqUrl:        MQ_URL,
	}
}

// 断开channel 和 connection
func (r *RabbitMQ) Destory() {
	r.ch.Close()
	r.conn.Close()
}

// 错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
	if err != nil {
		log.Fatalf("%s:%s", message, err)
		r.Destory()
		panic(fmt.Sprintf("%s:%s", message, err))
	}
}

// 创建RabbitMQ实例
// 简单模式下 queueName有值，后两个为空
// 路由模式和Topic模式下，queueName为空，后两个有值
// 订阅模式下 exchangeName 有值，其它两个值为空
func (r *RabbitMQ) NewRabbitMQ(queueName string, exchangeName string, routingKey string) *RabbitMQ {
	//创建RabbitMQ实例
	rabbitmq := NewRabbitMQStruct(queueName, exchangeName, routingKey)
	var err error
	// 获取connection
	rabbitmq.conn, err = amqp091.Dial(rabbitmq.MqUrl)
	rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
	// 获取channel
	rabbitmq.ch, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}

// 直接模式队列生产
func (r *RabbitMQ) PublishQueue(message string) {
	// 1.尝试创建队列
	_, err := r.ch.QueueDeclare(
		r.QueueName,
		// 是否持久化
		false,
		// 是否自动删除
		false,
		// 是否具有排他性
		false,
		// 是否阻塞处理
		false,
		// additional
		nil,
	)
	if err != nil {
		fmt.Println(err)
	}
	// 调用channel 发送消息到队列中
	r.ch.Publish(
		r.ExchangeName,
		r.QueueName,
		false, // 如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false, // 如果为true，当exchange 发送消息到队列后发现队列中没有消费者，则会把消息返还给发送者
		amqp091.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

// 通用交换机-发送消息
func (r *RabbitMQ) Publish(publish_type string, message string) {
	// 校验参数
	if !util.ContainsString([]string{"fanout", "direct", "topic"}, publish_type) {
		log.Fatal("publish_type is error")
	}
	// 创建交换机
	err := r.ch.ExchangeDeclare(
		r.ExchangeName,
		publish_type,
		true,  // 是否持久化
		false, // auto-deleted
		false, // 内部使用
		false, // no-wait
		nil,   // arguments
	)
	r.failOnErr(err, "Failed to declare an exchange")
	r.ch.Publish(
		r.ExchangeName,
		r.Key,
		false,
		false,
		amqp091.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

// queue消费者
func (r *RabbitMQ) ReceiveQueue() {
	// 1. 创建队列声明
	q, err := r.ch.QueueDeclare(
		r.QueueName,
		false, // 是否持久化
		false, // 是否自动删除
		false, // 是否具有排他性
		false, // 是否阻塞处理
		nil,   // 额外属性
	)
	if err != nil {
		fmt.Println(err)
	}
	// 2. 消费消息
	msgs, err := r.ch.Consume(
		q.Name, // queue
		"",     // consumer 用来区分多个消费者
		true,   // auto-ack
		false,  // 是否独有 exclusive
		false,  // no-local 设置为true 表示 不能将同一个Connection 中生产者传递给这个Connection 中的消费者
		false,  // no-wait 列是否阻塞
		nil,    // args
	)
	if err != nil {
		fmt.Println(err)
	}

	forever := make(chan bool)
	// 启用协和处理消息
	go func() {
		for d := range msgs {
			// 消息处理逻辑，可自行设计逻辑
			log.Printf("Received a message: %s", d.Body)
		}
	}()

	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}

// 交换机-消费端代码
// publish_type: fanout,direct,topic
// 路由模式使用 direct，话题模式使用 topic，
// 路由模式下，要注意key,规则
//
//	其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）
//	匹配 kuteng.* 表示匹配 kuteng.hello, kuteng.hello.one需要用kuteng.#才能匹配到
func (r *RabbitMQ) ReceiveExchange(publish_type string) {
	// 校验参数
	if !util.ContainsString([]string{"fanout", "direct", "topic"}, publish_type) {
		log.Fatal("publish_type is error")
	}
	// 试探性创建交换机
	err := r.ch.ExchangeDeclare(
		r.ExchangeName,
		publish_type, // 交换机类型
		true,
		false,
		false, //true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare an exchange")

	// 试探性创建队列，这里注意队列名称不要写
	q, err := r.ch.QueueDeclare(
		"", // 名称，不填的话，随机生成生效队列名称
		false,
		false,
		true,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to declare a queue")

	// 绑定队列到 exchange中
	err = r.ch.QueueBind(
		q.Name,
		r.Key, // routing key pub/sub 模式下，这里的key要是空字符串
		r.ExchangeName,
		false,
		nil,
	)
	r.failOnErr(err, "Failed to bind a queue to exchange")

	// 消费消息
	messages, err := r.ch.Consume(
		q.Name, // queue
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	r.failOnErr(err, "Failed to consume messages from queue")

	forever := make(chan bool)
	go func() {
		for d := range messages {
			log.Printf("Received message: %s", d.Body)
		}
	}()
	fmt.Println("Waiting for messages... 退出请按 CTRL+C")
	<-forever
}
